package com.example.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.example.behavior.service.BehaviorService;
import com.example.common.redis.CacheService;
import com.example.common.util.R;
import com.example.common.util.ResultCodeRecord;
import com.example.model.behavior.dto.LikesBehaviorDto;
import com.example.model.behavior.dto.ReadBehaviorDto;
import com.example.model.behavior.dto.UnLikesBehaviorDto;
import com.example.model.mess.UpdateArticleMess;
import com.example.model.user.pojo.User;
import com.example.utils.thread.UserThreadLocal;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import static com.example.common.constants.BehaviorConstants.LIKE_BEHAVIOR;
import static com.example.common.constants.BehaviorConstants.UN_LIKE_BEHAVIOR;
import static com.example.common.constants.HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;
import static com.example.model.mess.UpdateArticleMess.UpdateArticleType.LIKES;
import static com.example.model.mess.UpdateArticleMess.UpdateArticleType.VIEWS;

@Service
@Slf4j
public class BehaviorServiceImpl implements BehaviorService {
    @Autowired
    private CacheService cacheService;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public boolean like(LikesBehaviorDto likesBehaviorDto) {
        // 1、检查参数
        if (likesBehaviorDto == null || likesBehaviorDto.getArticleId() == null || checkParam(likesBehaviorDto)) {
            return R.error(ResultCodeRecord.PARAM_INVALID).isFlag();
        }
        // 2、是否登录
        User user = UserThreadLocal.getUser();
        if (user == null) {
            return R.error(ResultCodeRecord.NEED_LOGIN).isFlag();
        }
        UpdateArticleMess updateArticleMess = new UpdateArticleMess();
        updateArticleMess.setArticleId(likesBehaviorDto.getArticleId());
        updateArticleMess.setType(LIKES);

        // 3、点赞 保存数据
        if (likesBehaviorDto.getOperation() == 0) {
            Object object = cacheService.hGet(LIKE_BEHAVIOR + likesBehaviorDto.getArticleId(), user.getUserId().toString());
            if (object != null) {
                log.info("该用户：{}已经点赞过，直接返回成功", user.getUserId());
                return R.error(ResultCodeRecord.PARAM_INVALID, "已点赞").isFlag();
            }
            // 保存当前key
            log.info("保存当前key:{}, {}, {}", likesBehaviorDto.getArticleId() , user.getUserId(), likesBehaviorDto);
            cacheService.hPut(LIKE_BEHAVIOR + likesBehaviorDto.getArticleId(), user.getUserId().toString(), JSON.toJSONString(likesBehaviorDto));
            updateArticleMess.setAdd(1);
        } else {
            // 删除当前key
            log.info("删除当前key:{}, {}", likesBehaviorDto.getArticleId() , user.getUserId());
            cacheService.hDelete(LIKE_BEHAVIOR + likesBehaviorDto.getArticleId(), user.getUserId().toString());
            updateArticleMess.setAdd(-1);
        }
        // 发送消息，数据聚合
        kafkaTemplate.send(HOT_ARTICLE_SCORE_TOPIC, JSON.toJSONString(updateArticleMess));
        // 4、返回结果
        return true;
    }

    @Override
    public boolean unLike(UnLikesBehaviorDto unLikesBehaviorDto) {
        if (unLikesBehaviorDto.getArticleId() == null) {
            return R.error(ResultCodeRecord.PARAM_INVALID).isFlag();
        }
        User user = UserThreadLocal.getUser();
        if (user == null) {
            return R.error(ResultCodeRecord.NEED_LOGIN).isFlag();
        }
        if (unLikesBehaviorDto.getType() == 0) {
            log.info("保存当前key:{} ,{}, {}", unLikesBehaviorDto.getArticleId(), user.getUserId(), unLikesBehaviorDto);
            log.info("该用户：{}已经点赞过不喜欢，直接返回成功", user.getUserId());
            cacheService.hPut(UN_LIKE_BEHAVIOR + unLikesBehaviorDto.getArticleId().toString(), user.getUserId().toString(), JSON.toJSONString(unLikesBehaviorDto));
        } else {
            log.info("删除当前key:{} ,{}, {}", unLikesBehaviorDto.getArticleId(), user.getUserId(), unLikesBehaviorDto);
            cacheService.hDelete(UN_LIKE_BEHAVIOR + unLikesBehaviorDto.getArticleId().toString(), user.getUserId().toString());
        }
        return true;
    }

    @Override
    public boolean read(ReadBehaviorDto readBehaviorDto) {
        // 1、检查参数
        if (readBehaviorDto == null || readBehaviorDto.getArticleId() == null) {
            return R.error(ResultCodeRecord.PARAM_INVALID).isFlag();
        }
        // 2、是否登录
        User user = UserThreadLocal.getUser();
        if (user == null) {
            return R.error(ResultCodeRecord.NEED_LOGIN).isFlag();
        }
        // 更新阅读次数
        String readBehaviorJson = (String) cacheService.hGet(LIKE_BEHAVIOR + readBehaviorDto.getArticleId(), user.getUserId().toString());
        if (StringUtils.isNotBlank(readBehaviorJson)) {
            ReadBehaviorDto readBehaviorDtos = JSON.parseObject(readBehaviorJson, ReadBehaviorDto.class);
            readBehaviorDto.setCount((short) (readBehaviorDtos.getCount() + readBehaviorDto.getCount()));
        }
        // 保存当前key
        log.info("保存当前key:{}, {}, {}", readBehaviorDto.getArticleId() , user.getUserId(), readBehaviorDto);
        cacheService.hPut(LIKE_BEHAVIOR + readBehaviorDto.getArticleId(), user.getUserId().toString(), JSON.toJSONString(readBehaviorDto));

        // 发送消息，数据聚合
        UpdateArticleMess updateArticleMess = new UpdateArticleMess();
        updateArticleMess.setArticleId(readBehaviorDto.getArticleId());
        updateArticleMess.setType(VIEWS);
        updateArticleMess.setAdd(1);
        kafkaTemplate.send(HOT_ARTICLE_SCORE_TOPIC, JSON.toJSONString(updateArticleMess));
        return true;
    }

    /**
     * 检查参数
     *
     * @return
     */
    private boolean checkParam(LikesBehaviorDto dto) {
        if (dto.getType() > 2 || dto.getType() < 0 || dto.getOperation() > 1 || dto.getOperation() < 0) {
            return true;
        }
        return false;
    }
}
